/*******************************************************************************
* COPYRIGHT LIPUJUN TECH CO.,LTD
********************************************************************************
* 文件名称:  mqttMsmgCmd.c
* 功能描述:
* 使用说明:  无
* 文件作者:  LIPUJUN
* 编写日期:  2018/11/28
* 修改历史:
* 修改版本  修改日期       修改人   修改内容
* ------------------------------------------------------------------------------
  01a       2018/11/28       李普军   创建
*******************************************************************************/

/******************************* 包含文件声明 ********************************/
#include "user_core.h"
#include "link.h"
#include <sys/vfs.h>
/* 消息 处理 */
AIR_APP_INFO_T stMqttMsgInfo;
VER_UPDATE_INFO_T stVerUpdate;
CHAR g_rsyncCmd[0x400] = {0};
CHAR g_upgradeRet[0x40] = {0};

INT32 mqttRegHostMsg ();
INT32 mqttRegHostMsg_ACK(CHAR *buf, UINT32 len);
INT32 mqttDataSyncStart (CHAR *buf, UINT32 len);
INT32 mqttDataSyncDone (CHAR *buf, UINT32 len);
INT32 mqttCleanHdisk (CHAR *buf, UINT32 len);

INT32 verUpdate (void);

INT32 mqttFirmwareUpdate (CHAR *buf, UINT32 len);
INT32 mqttFirmwareUpdateDone (CHAR *result);
INT32 mqttFirmwareUpdateDone_ACK (CHAR *buf, UINT32 len);

INT32 mqtt_HB_MSG_ACK (CHAR *buf, UINT32 len);
INT32 mqttMsgRegHostRule (void);

INT32 onoffTime_ACK(CHAR *buf, UINT32 len);



MQTT_MSG_PROCESS_CMD_T g_stMsgCmdList[] =
{
    {CMD_REG             ,  mqttRegHostMsg_ACK         ,  "mqttRegHostMsg_ACK"          },//注册
    {CMD_UPLOAD_FILE     ,  mqttDataSyncStart          ,  "mqttDataSyncStart"           },//上传文件
    {CMD_UPLOAD_RET      ,  mqttDataSyncDone           ,  "mqttDataSyncDone"            },//上传结果
    {CMD_DISK_CLEAR      ,  mqttCleanHdisk             ,  "mqttCleanHdisk"              },//清空硬盘
    {CMD_UPGRADE_CMD     ,  mqttFirmwareUpdate         ,  "mqttFirmwareUpdate"          },//更新
    {CMD_UPGRADE_RET	 ,  mqttFirmwareUpdateDone_ACK ,  "mqttFirmwareUpdateDone_ACK"  },//更新结果
    {CMD_HEART_BEAT      ,  mqtt_HB_MSG_ACK            ,  "mqtt_HB_MSG_ACK"             },//心跳
    {CMD_ONOFF           ,  onoffTime_ACK              ,  "onoffTime_ACK"               }, //
    {CMD_NULL            ,  NULL                       ,  ""                            }
};

INT32 user_init (void)
{
    INT32 ret = 0;

    memset (&stVerUpdate, 0x00, sizeof(VER_UPDATE_INFO_T));
    stMqttMsgInfo.rData.data = (CHAR *)malloc (MSG_QUEUE_DATA_LEN);
    if (NULL == stMqttMsgInfo.rData.data)
    {
         Error ("stMqttMsgInfo.rData.data, Malloc Fail !");
         return ERROR;
    }

    stMqttMsgInfo.sData.data= (CHAR *)malloc (MSG_QUEUE_DATA_LEN);
    if (NULL == stMqttMsgInfo.sData.data)
    {
         Error ("stMqttMsgInfo.sData.data, Malloc Fail !");
	 free(stMqttMsgInfo.rData.data);
         return ERROR;
    }

    ret = sem_init (&stMqttMsgInfo.msgSem, 0, 1);
    if(0 != ret)
    {
        Error ("stMqttMsgInfo.msgSem, Semaphore Init Fail !");
	free(stMqttMsgInfo.rData.data);
	free(stMqttMsgInfo.sData.data);
        return ERROR;
    }

    ret = sem_init (&stMqttMsgInfo.timeSem, 0, 0);
    if(0 != ret)
    {
        Error ("stMqttMsgInfo.timeSem, Semaphore Init Fail !");
	free(stMqttMsgInfo.rData.data);
	free(stMqttMsgInfo.sData.data);
        sem_destroy (&stMqttMsgInfo.msgSem);
        return ERROR;
    }

    ret = sem_init (&stMqttMsgInfo.uploadSem, 0, 0);
    if(0 != ret)
    {
        Error ("stMqttMsgInfo.uploadSem, Semaphore Init Fail !");
	free(stMqttMsgInfo.rData.data);
	free(stMqttMsgInfo.sData.data);
	sem_destroy (&stMqttMsgInfo.msgSem);
	sem_destroy (&stMqttMsgInfo.timeSem);
        return ERROR;
    }
    

    ret = sem_init (&stMqttMsgInfo.upgradeSem, 0, 0);
    if(0 != ret)
    {
         Error ("stMqttMsgInfo.upgradeSem, Semaphore Init Fail !");
	free(stMqttMsgInfo.rData.data);
	free(stMqttMsgInfo.sData.data);
	sem_destroy (&stMqttMsgInfo.msgSem);
	sem_destroy (&stMqttMsgInfo.timeSem);
	sem_destroy (&stMqttMsgInfo.uploadSem);
         return ERROR;
    }   

    return OK;
}
INT32 getID(const CHAR *data)
{
    if(data == NULL || strlen(data) <= 0)
        return -1;

    while(*data != '_' && *data != '\0')
        data++;
    if(*data != '_' || *data == '\0')
        return -1;
    data++;
    return atoi(data);   
}
UINT32 getProfileConf(const char *key)
{
    CHAR filename[512] = {0};
    CHAR data[LINE_CONTENT_MAX_LEN]={0};
    
    if(key == NULL)
        return 0;
    sprintf(filename,"%s/%s",stMqttMsgInfo.baseUploadPath,VIDEO_CONF);  
    pthread_rwlock_rdlock(&stMqttMsgInfo.profile_lock); 
    com_getKeyValue(key ,data, filename);   
    pthread_rwlock_unlock(&stMqttMsgInfo.profile_lock);
    return atoi(data);
}
void setProfileConf(const char *key ,UINT32 value)
{
    CHAR filename[512] = {0};
    CHAR data[LINE_CONTENT_MAX_LEN]={0};
    
    if(key == NULL)
        return;
    sprintf(filename,"%s/%s",stMqttMsgInfo.baseUploadPath,VIDEO_CONF);   
    sprintf(data,"%d",value);
   
    pthread_rwlock_wrlock(&stMqttMsgInfo.profile_lock);
    com_setKeyValue(key ,data, filename);
    pthread_rwlock_unlock(&stMqttMsgInfo.profile_lock);
}
int isGroup(void)
{
    int index[BLOCK_MAX_COUNT] = {0};
    
    if(sort_dir(stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,index) > 0)
        return index[0];
    return -1;
}
int isBlock(int groupId)
{
    int index[BLOCK_MAX_COUNT] = {0};
    char path[1024] = {0};

    if(groupId < 0)
        return -1;
    sprintf(path,"%s/%s%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP,groupId);
    if(sort_dir(path,VIDEO_BLOCK ,index) > 0)
        return index[0];
    return -1;
}
int getBlockSizeInBase(void)
{
    int index[BLOCK_MAX_COUNT] = {0};
    
    return sort_dir(stMqttMsgInfo.baseUploadPath,VIDEO_BLOCK ,index);
}
int getGroupCount(void)
{
    int index[BLOCK_MAX_COUNT] = {0};
    
    return sort_dir(stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,index);    
}
int getBlockCount(int groupId)
{
    int index[BLOCK_MAX_COUNT] = {0};
    char path[1024] = {0};

    if(groupId < 0)
        return -1;
    sprintf(path,"%s/%s%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP,groupId);
    return sort_dir(path,VIDEO_BLOCK ,index);
}
char* getBlockID(int groupid,char *value)
{
    char filename[512] = {0};
    char ID[LINE_CONTENT_MAX_LEN] = {0};

    if(groupid < 0 || value == NULL)
	return "";

    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupid,VIDEO_CONF);
    com_getKeyValue("block_ID", ID, filename);
    strcpy(value,ID);
    return value;    
}
char* getGroupID(int groupid,char *value)
{
    char filename[512] = {0};
    char ID[LINE_CONTENT_MAX_LEN] = {0};

    if(groupid < 0 || value == NULL){
	return "";
    }
    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupid,VIDEO_CONF);

    com_getKeyValue("group_ID", ID, filename);
    strcpy(value,ID);
    return value;
}
void setBlockID(int groupid,const char *value)
{
    INT8 filename[512] = {0};

    if(groupid < 0 || value == NULL)
	return;

    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupid,VIDEO_CONF);
    //htkj_system("echo 'block_ID=%s' >> %s",value,filename);
    if(access(filename,F_OK) != 0)
        htkj_system("touch %s",filename);
    com_setKeyValue("block_ID" ,value,filename);
}

void setGroupID(int groupId,const char *value)
{
    INT8 filename[512] = {0};

    if(groupId < 0 || value == NULL)
	return;

    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);
    //htkj_system("echo 'group_ID=%s' > %s",value,filename);
    if(access(filename,F_OK) != 0)
        htkj_system("touch %s",filename);
    com_setKeyValue("group_ID" ,value,filename);
}
UINT32 getBlockDataSize(const char *groupId,const char * blockId)
{
    INT8 filename[512] = {0};
    char key[LINE_CONTENT_MAX_LEN] = {0};
    char value[LINE_CONTENT_MAX_LEN] = {0};
    int gid = getID(groupId) ;
    int bid = getID(blockId); 
    int size = 0;

    if(gid < 0 || bid < 0)
	return 0;

    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,gid,VIDEO_CONF);
    if(access(filename, F_OK) != 0){
        sprintf(filename,"%s/%s%d/block%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,gid,bid);
        return getPathDataSize(filename);
    }
    sprintf(key,"block%dSize",bid);
    com_getKeyValue(key, value, filename);
    size = atoi(value);
    if(size <= 0){
        sprintf(filename,"%s/%s%d/block%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,gid,bid);
        return getPathDataSize(filename);
    }
    return size;//kB
}
UINT32 getGroupSize(int groupId)
{
    INT8 filename[512] = {0};
    int index[BLOCK_MAX_COUNT] = {0};
    UINT32 sumSize = 0;
    char key[LINE_CONTENT_MAX_LEN] = {0};
    char value[LINE_CONTENT_MAX_LEN] = {0};
    int size = 0;
    int i = 0;

    if(groupId < 0)
	return 0;

    sprintf(filename,"%s/%s%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP,groupId);
    size = sort_dir(filename,VIDEO_BLOCK ,index);

    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);
    if(access(filename, F_OK) != 0){
        sprintf(filename,"%s/%s%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId);
        return getPathDataSize(filename);
    }
    
    for(i = 0 ; i < size ; i++){
        memset(value,0,sizeof(value));
        sprintf(key,"block%dSize",index[i]);//i
        com_getKeyValue(key, value, filename);
        sumSize += atoi(value);//kB
    }
    return sumSize;//KB

   // return getPathData_realSize(filename);
}
INT32 getBlockCountByConf(int groupId)
{
    CHAR filename[512] = {0};
    INT32 count = 0;
    //INT32 checkCount = getBlockCountByConf(groupId);
    CHAR key[64] = {0};
    CHAR value[64] = {0};
    INT32 size = 0;
    INT32 i = 0;

    if(groupId < 0)
        return 0;    
    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);
    if(access(filename, F_OK) != 0){
        sprintf(filename,"%s/%s%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId);
        return 0;
    }
    sprintf(filename,"cat %s/%s%d/%s | wc -l",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);
    systemOut ( filename , value, sizeof(value));//"/tmp/.groupline1");
    size = atoi(value);
    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);
    for(i = 0 ; i < size ; i++){
        memset(value,0,sizeof(value));
        sprintf(key,"block%dSize",i);
        com_getKeyValue(key, value, filename);
        if(atoi(value) > 0)
            count++;
    }
    return count;    
}
void setGroupInfo(int groupId)
{
    INT8 filename[512] = {0};
    INT8 count[10] = {0};

    if(groupId < 0)
	return;

    sprintf(count,"%d",getBlockCount(groupId));
    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);
    //htkj_system("echo 'group_ID=%s' > %s",value,filename);
    if(access(filename,F_OK) != 0)
        htkj_system("touch %s",filename);
    com_setKeyValue("block_count" ,count,filename);
    sprintf(count,"%d",getGroupSize(groupId));
    com_setKeyValue("AllSize" ,count,filename);
}
UINT32 getBlocksCountInGroup(int groupId)
{
    char filename[512] = {0};
    char count[LINE_CONTENT_MAX_LEN] = {0};
    UINT32 size = 0;
    UINT32 nSize = 0;

    if(groupId < 0)
	return 0;

    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);
    com_getKeyValue("block_count", count, filename);
    
    size = atoi(count);
    nSize = getBlockCountByConf(groupId);
    if(nSize > size){       
        if(access(filename,F_OK) != 0)
            htkj_system("touch %s",filename);           
        sprintf(count,"%d",nSize);
        com_setKeyValue("block_count" ,count,filename);
        size = nSize;
    }   
    return size;     
}

UINT32 getGroupOriginalSize(int groupId)
{
    char filename[512] = {0};
    char count[LINE_CONTENT_MAX_LEN] = {0};
    UINT32 size = 0;
    UINT32 nSize = 0;

    if(groupId < 0)
	return 0;

    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);
    com_getKeyValue("AllSize", count, filename);
    
    size = atoi(count);
    nSize = getGroupSize(groupId);
    if(nSize > size){       
        if(access(filename,F_OK) != 0)
            htkj_system("touch %s",filename);           
        sprintf(count,"%d",nSize);
        com_setKeyValue("AllSize" ,count,filename);
        size = nSize;
    }   
    return size;   
}

void clearBlockID(int groupId)
{
    INT8 filename[512] = {0};

    if(groupId < 0)
	return;

    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);

    com_rmKeyValue("block_ID", filename);
}
void clearGroupID(int groupId)
{    
    if(groupId < 0)
	return;

    htkj_system("rm -f %s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_CONF);
}
void clearGroupData(int groupId)
{
    if(groupId < 0)
	return;

    htkj_system("rm -fr %s/%s%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId);
}

UINT32 getAllBlockCount(void)
{
    INT32 size = -1;
    INT32 i = 0;
    INT32 index[BLOCK_MAX_COUNT] = {0};
    INT32 count = 0;

    size = sort_dir(stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,index);
    
    if(size <= 0){
        return 0;
    }
    for(i = 0; i < size;i++){
        int tsize = getBlockCount(index[i]);
        if(tsize == 0){
            clearGroupID(index[i]);
            continue;
        }
        count += tsize;
    }
    delEmptyDir(stMqttMsgInfo.baseUploadPath);
    return count;    
}
INT32 uploadBlockDataSize()
{
    INT32 size = 0;
    INT32 blockSize = 0;
    INT32 i = 0;
    INT32 index[BLOCK_MAX_COUNT] = {0};
    CHAR  path[512] = {0};
    
    blockSize = sort_dir(stMqttMsgInfo.baseUploadPath,VIDEO_BLOCK ,index);
    for(i = 0;i < blockSize ; i++){
        sprintf(path,"%s/%s%d" ,stMqttMsgInfo.baseUploadPath ,VIDEO_BLOCK,index[i]);
        size += getPathDataSize(path);       
    }
    return size;    
}
void check_divideGroup(void)
{
    UINT32 curTi = currentTimeGet_int();    
    UINT32 lastTi = getProfileConf(lastNewGroupTime);
   
    if(curTi > lastTi && (curTi - lastTi) > (stMqttMsgInfo.route_min_time*60))
    {   
        pthread_mutex_lock(&stMqttMsgInfo.block_mutex);
       // if(getPathFilesCount(stMqttMsgInfo.uploadLinkPath) > 1){
            divide_block();            
       // }                
        if(getBlockSizeInBase() > 0){
            if(uploadBlockDataSize()/1024.0/1024.0 >= GROUP_MIN_SIZE){                       
                setGroupInfo(divide_group());//tmp new group record block count            
               // logfile_prepare();
            } 
        }
        pthread_mutex_unlock(&stMqttMsgInfo.block_mutex);
        setProfileConf(lastNewGroupTime ,curTi);       
    }
    
}

/*******************************************************************************
*   函数名称: regHostMsg
*   功    能: 注册 HOST 消息
*   函数类型:
*   参    数:
*   参数名称       类  型         输入/输出         描述
*   buf             CHR *           IN              数据
*   len             UINT32          IN              数据长度
*   函数返回: OK 或失败的错误码
*   说    明:
*******************************************************************************/
INT32 mqttRegHostMsg ()
{
    INT32  ret = 0;
    CHAR   sBuf[0x100] = {0};
    CHAR   mac[LINE_CONTENT_MAX_LEN] = {0};
    UINT32 uploadSize = 0;//tmp
    INT32 videoaddSize = 0;

    check_divideGroup();
    uploadSize = getPathDataSize(stMqttMsgInfo.baseUploadPath);//tmp
    videoaddSize = uploadSize - getProfileConf(lastVideoSize);
    if(videoaddSize < 0)
        videoaddSize = 0;  

    get_localMAC(mac,stMqttMsgInfo.netName);   
    sprintf (sBuf, "%s,%s,%s,%.2f,%.2f,%d,%d", stMqttMsgInfo.appVer, mac, stMqttMsgInfo.ip,
		(double)uploadSize*1.0/1024,(double)videoaddSize*1.0/1024, stMqttMsgInfo.UplugTimes + getUSBDiskPlugTimes(0),(stMqttMsgInfo.oNTimeDuration + 30)/60);
    
    ret = air_send_msg(CMD_REG ,sBuf, strlen(sBuf),0);
    return (ret);
}


INT32 mqttMsgRegHostRule (void)
{
    if ((stMqttMsgInfo.regHostAck == MQTT_REG_HOST_MSG_INIT) && (stMqttMsgInfo.link == IOT_STATUS_CONNECT))
    {

        mqttRegHostMsg ();
    }
    
    return OK;
}

/*******************************************************************************
*   函数名称: regHostMsg_ACK
*   功    能: HOST 应答
*   函数类型:
*   参    数:
*   参数名称       类  型         输入/输出         描述
*   buf             CHR *           IN              数据
*   len             UINT32          IN              数据长度
*   函数返回: OK 或失败的错误码
*   说    明:
*******************************************************************************/
void loadRegTime(const char* sRTime)//settime
{
    UINT32 RTi = strTime2Int(sRTime);
    UINT32 localTi = currentTimeGet_int();    
    INT8 localTs[0x40] = {0};
    INT8 startTs[0x40] = {0};

    if(access(REGSub6ATimeFlg,F_OK) != 0){
        if(RTi > 0 && localTi > 0 && (RTi > localTi) && ((RTi - localTi) > 10) && ((RTi - localTi) < 1800)){
            htkj_system("echo regSub6ATime=%d > %s",RTi - localTi + 15,REGSub6ATimeFlg);
             INFO("no need load reg time:[%d > %d]" ,localTi ,RTi);
        }
    }
    stMqttMsgInfo.timeChange = 1;
    htkj_system ("date -s '%s'", sRTime);
    intTime2Str(localTi ,localTs);
    stMqttMsgInfo.powerONTime = RTi - (localTi - stMqttMsgInfo.powerONTime);    
    intTime2Str(stMqttMsgInfo.powerONTime ,startTs);
    localTi = currentTimeGet_int();

    INFO("load reg time:[%s;%d;%s=%s]",startTs ,localTi - stMqttMsgInfo.powerONTime, localTs, sRTime);    
}
void logfile_prepare()
{
    char path[512] = {0};
    int groupid = isGroup();
    int blockid = -1;
   
    if(groupid >= 0)
        blockid = isBlock(groupid);   
    
    if(groupid >= 0 && blockid >= 0){
        sprintf(path,"%s/%s%d/%s%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP,groupid,VIDEO_BLOCK,blockid);
        log_upload_prepare(path);//cp log to first group/first block
    }
}
INT32 mqttRegHostMsg_ACK(CHAR *buf, UINT32 len)
{   
    CHAR date[0x40] = {0};
    CHAR result[0x40] = {0};
    CHAR isFw[0x40] = {0};
    CHAR block_MAX_size[0x40] = {0};
    CHAR route_MIN_time[0x40] = {0};
    INT32 blockSize = 0;
    INT32 routeTime = 0;

    if (REG_HOST_ACK_SUM == sscanf(buf, "%[^,],%[^,],%[^,],%[^,],%[^;]" ,date ,block_MAX_size ,route_MIN_time ,isFw ,result))
    {
        if (0 != strcmp (result, "ok"))
        {
            Error ("Reg Host Fail!");
	    return ERROR;
        }
        stMqttMsgInfo.UplugTimes = 0;          
        stMqttMsgInfo.oNTimeDuration = 0;
        usleep(10);
        setProfileConf(lastPowerTimeLen ,0);
        setProfileConf(lastVideoSize,getPathDataSize(stMqttMsgInfo.baseUploadPath));
	loadRegTime(date);
        if(getProfileConf(lastNewGroupTime) == 0){//init data/.profile
             setProfileConf(lastNewGroupTime ,currentTimeGet_int());
        }
        blockSize = atoi(block_MAX_size);
        routeTime = atoi(route_MIN_time);
        stMqttMsgInfo.isHaveFwUpgrade = atoi(isFw);
        stMqttMsgInfo.regHostAck = MQTT_REG_HOST_MSG_OK;
        stMqttMsgInfo.HBAckTime = currentTimeGet_int();	
        if(blockSize > 0 && blockSize <= BLOCK_MAX_COUNT && blockSize != stMqttMsgInfo.block_max_size){
            com_setKeyValue("block_max_size",block_MAX_size,stMqttMsgInfo.appConfig);//set to file
            stMqttMsgInfo.block_max_size = blockSize; 
        }
        if(routeTime > 0 && routeTime != stMqttMsgInfo.route_min_time){
            com_setKeyValue("route_min_time",route_MIN_time,stMqttMsgInfo.appConfig);//set to file
            stMqttMsgInfo.route_min_time = routeTime;
        }
        INFO ("register server OK");
        //divide group
        //check_divideGroup();
    }

    return OK;
}
//0:ok,1:uploading data;2:no block for uploading
INT32 getFirstValidBlock(CHAR *groupID,CHAR *blockID ,CHAR *path)
{
    INT32 gid = getID(groupID);
    INT32 bid = getID(blockID);
    CHAR buf[0x40] = {0};
    CHAR value[0x40] = {0};
    //UINT32 blockCount = getAllBlockCount();

    if(stMqttMsgInfo.dataSync == 1){
        return 1;
    }
    if(gid >= 0 && bid >= 0){//xu chuan
        if(strcmp(groupID,getGroupID(gid,buf)) == 0){
            if(isBlock(gid) == bid){
                sprintf(path,"%s/%s%d/%s%d",stMqttMsgInfo.baseUploadPath ,VIDEO_GROUP ,gid ,VIDEO_BLOCK ,bid);
                return 0;
            }
        }
    }

    if(gid < 0 || bid < 0){
        while(1){
            if((gid = isGroup()) < 0)//Invalid group
		return 2;           
            if((bid = isBlock(gid)) < 0){//Invalid block
                clearGroupData(gid);
                continue;
            }else
                break;              
        }                 
    }

    if(gid < 0 || bid < 0)
        return 2;

    if (gid != getID(getGroupID(gid,buf))){//new groupid
        sprintf(groupID,"%s_%d_%d",stMqttMsgInfo.devId ,gid ,currentTimeGet_int());   
    }else{
        getGroupID(gid,groupID);
    }
    setGroupID(gid,groupID);
    getBlockID(gid ,value);
    if(strlen(value) > 0){
        sprintf(blockID,"%s",value);
    } else {
        sprintf(blockID,"%s_%d_%d",stMqttMsgInfo.devId ,bid ,currentTimeGet_int());//new blockid
        setBlockID(gid,blockID); 
    }
    sprintf(path,"%s/%s%d/%s%d" ,stMqttMsgInfo.baseUploadPath ,VIDEO_GROUP ,gid ,VIDEO_BLOCK , bid);
  
    return 0;    
}
/*******************************************************************************
*   函数名称: dataSyncStart
*   功    能: HOST 应答
*   函数类型:
*   参    数:
*   参数名称       类  型         输入/输出         描述
*   buf             CHR *           IN              数据
*   len             UINT32          IN              数据长度
*   函数返回: OK 或失败的错误码
*   说    明:
*******************************************************************************/
void uploadLogs(CHAR *buf,UINT32 len)
{
    CHAR ip[0x40] = {0};
    CHAR user[0x40] = {0};
    CHAR pw[0x40] = {0};
    CHAR remotePath[0x80] = {0};
    CHAR groupID[0x40] = {0};
    CHAR blockID[0x40] = {0};
    CHAR path[0x200] = {0};

    if (DATA_SYNC_START_SUM == sscanf(buf, "%[^,],%[^,],%[^,],%[^,],%[^,],%[^;]", ip, user, pw, remotePath ,groupID ,blockID))
    { 
       sprintf(path,"%s/log/upload",DATA_STORE_BASE_PATH);
       log_upload_prepare(path);//cp log to first group/first block
       INFO ("==================ftp start upload %s",path);
       htkj_system ("lftp -u %s,%s -e 'set file:charset utf-8 ; set xfer:clobber on ; mirror --Remove-source-files -R -c --parallel=5 %s/. log/%s/ ; exit ;' %s && echo 'logs upload complete' >> %s 2>&1 &", user, pw, path ,remotePath, ip,SHELL_LOG);       
    }
}
void check_blockConf(int groupId)
{
    CHAR filename[512] = {0};
    CHAR key[64] = {0};
    CHAR value[64] = {0};
    INT32 bIndex[BLOCK_MAX_COUNT] = {0};     
    CHAR path[512] = {0};
    INT32 size = 0;
    INT32 i = 0;
    INT32 flg = 0;

    if(groupId < 0)
        return;    
    sprintf(filename,"%s/%s%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId);  
    size = sort_dir(filename ,VIDEO_BLOCK ,bIndex);
    sprintf(filename,"%s/%s%d/%s",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP,groupId ,VIDEO_CONF);  
    for(i = 0; i < size;i++){
        INT32 cSize = 0;
        memset(value,0,sizeof(value));
        sprintf(key,"block%dSize",bIndex[i]);
        com_getKeyValue(key, value, filename);
        cSize = atoi(value);
        if(cSize <= 0){
            sprintf(path,"%s/%s%d/%s%d",stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,groupId,VIDEO_BLOCK,bIndex[i]);
            cSize = getPathData_realSize(path);
            sprintf(value,"%d",cSize);
            com_setKeyValue(key,value,filename);
            flg = 1;
        }
    }
    if(flg == 1)
         setGroupInfo(groupId);
    
}
INT32 mqttDataSyncStart (CHAR *buf, UINT32 len)
{    
    CHAR ip[0x40] = {0};
    CHAR user[0x40] = {0};
    CHAR pw[0x40] = {0};
    CHAR remotePath[0x80] = {0};
    CHAR result[0x40] = {0};
    UINT32 fileCount = 0;
    UINT32 totalSize = 0;
    CHAR starttime[0x40] = {0};
    CHAR groupID[0x40] = {0};
    CHAR blockID[0x40] = {0};  
    CHAR sBuf[0x400] = {0};
    CHAR path[0x200] = {0};
    CHAR needStartftp = 0;

    if (DATA_SYNC_START_SUM == sscanf(buf, "%[^,],%[^,],%[^,],%[^,],%[^,],%[^;]", ip, user, pw, remotePath ,groupID ,blockID))
    {
        uploadLogs(buf,len);
        switch(getFirstValidBlock(groupID ,blockID ,path)){
            case 0:
                htkj_system("chmod 777 %s -R",path);
                memset (g_rsyncCmd, 0x00, sizeof(g_rsyncCmd));  
	        sprintf (g_rsyncCmd, "lftp -u %s,%s -e 'set file:charset utf-8 ; set xfer:clobber on ; mirror -R -c --parallel=%d %s/. %s/ ; exit ;' %s >> %s 2>&1 &", user, pw,stMqttMsgInfo.ftpChannelNum, path ,remotePath, ip,SHELL_LOG);
                strcpy(result,"ok");
                INFO ("FTP command: %s", g_rsyncCmd);
                fileCount = getPathFilesCount(path);
	        //totalSize = getPathData_realSize(path);//KB
                totalSize = getBlockDataSize(groupID ,blockID);
                if(totalSize <= 0 ){
                   totalSize = getPathData_realSize(path);
                }
                if(totalSize < VIDEO_BLOCK_MIN_SIZE)
                    strcpy(result,"nodata");
                else  {
                    strcpy(result,"ok");
                    needStartftp = 1;
                }
            break;
            case 1:
                htkj_system("chmod 777 %s -R",path);                
                fileCount = getPathFilesCount(path);
	        //totalSize = getPathData_realSize(path);//KB
                totalSize = getBlockDataSize(groupID ,blockID);
                if(totalSize <= 0 ){
                   totalSize = getPathData_realSize(path);
                }
                if(totalSize < VIDEO_BLOCK_MIN_SIZE)
                    strcpy(result,"nodata");
                else  
                    strcpy(result,"ok");
                INFO ("ftp (started) upload data");
            break;
            case 2:
               strcpy(result,"nodata");
               INFO ("ftp no data for uploading");
            break;
            default:
               strcpy(result,"nodata");
            break;
        }      
        
    }else{
        strcpy (result, "Parse error");
    }
    
    strcpy (starttime, currentTimeGet()); 
    if(strcmp(result,"ok") != 0){
        strcpy(groupID,"0");
        strcpy(blockID,"0");
    }
    stMqttMsgInfo.curGroupID = getID(groupID);
    stMqttMsgInfo.curBlockID = getID(blockID);
    sprintf (sBuf, "%d,%.2f,%s,%s,%s,%d,%d,%.2f,%s" ,fileCount ,(double)totalSize/1024 ,starttime ,groupID ,blockID ,getID(blockID)+1 ,getBlocksCountInGroup(getID(groupID)) ,(double)getGroupOriginalSize(getID(groupID))/1024.0,result);//MB //tmp  block_num   blocks_count   group_size
 
    if(needStartftp){
        INFO("ftp upload file prestart %s",sBuf);
	sem_post (&stMqttMsgInfo.uploadSem);
    }
//printf("test msg:%s\n",sBuf);
    air_send_msg(CMD_UPLOAD_FILE ,sBuf, strlen(sBuf), 1);
    check_blockConf(stMqttMsgInfo.curGroupID);
    return OK;
}
//0 no lftp and not send;1 no ftp sending;2 lftp sending;3 lftp not sending
INT32 net_speed()
{
    CHAR result[64] = {0};
    systemOut("pgrep lftp" ,result, sizeof(result));//"/tmp/.lftpexit.tmp");
    if(atoi(result) <= 0){//no lftp	
        return 0;
    }
    
    if(stMqttMsgInfo.speed > 1){//speed > 1kB 	 lftp sending
	return 1;
    }
    return 2;//lftp maybe send data not send data
}


INT32 dataFileSync_task (void *arg)
{
    while (1)
    {
	char flg = 0;	       
        UINT32 sendingTime = 0;

        sem_wait (&stMqttMsgInfo.uploadSem);
	stMqttMsgInfo.dataSync = 1;
        INFO ("==================ftp start upload DATA_SYNC_START...");       
 	system (g_rsyncCmd);	
	sleep(2);
        sendingTime = currentTimeGet_int();
	while(1){
            INT32 ret = 0;
            UINT32 curTime = 0;
	    usleep(200000);//200ms
	    ret = net_speed();
            curTime = currentTimeGet_int();
	    if(ret == 0){//no ftp ,speed < 1kB		
		flg = 1;//lftp upload ok
                stMqttMsgInfo.dataSync = 0; 
		break;
	    }else if(ret == 1){//no ftp ,speed >= 1kB
                sendingTime = curTime;
                continue;
            }else if(ret == 2){//lftp no speed
                if(stMqttMsgInfo.regHostAck == MQTT_REG_HOST_MSG_INIT){//need reg                   
		    system("killall -9 lftp > /dev/null 2>&1");                  
 		    Error ("ftp upload file FAIL stMqttMsgInfo.regHostAck == MQTT_REG_HOST_MSG_INIT");	    	    
                    stMqttMsgInfo.dataSync = 0;                
		    break;
                }else if((curTime - sendingTime) > stMqttMsgInfo.ftpNoTimeout){// timeout 10min
                    system("killall -9 lftp > /dev/null 2>&1");			    
                    sem_post (&stMqttMsgInfo.uploadSem);
                    Error ("ftp upload file FAIL restart ftp");                                        
		    break;
                }
            }            

        }	
	if(flg > 0){	 
	    INFO ("ftp upload file end SUCCESS");
	}else{	      
            Error ("ftp upload file end FAIL");
	}
    }
    return OK;
}

INT32 real_cleanData(const char *groupID ,const char *blockID)
{
    CHAR path[512] = {0};
    int gid = getID(groupID);
    int bid = getID(blockID);

    if(gid < 0 || bid < 0){
        return -1;
    }
    if(strcmp(groupID ,getGroupID(gid ,path)) != 0){
        return -1;
    }
 
    htkj_system("rm -fr %s/%s%d/%s%d" ,stMqttMsgInfo.baseUploadPath ,VIDEO_GROUP ,gid ,VIDEO_BLOCK ,bid);
    
    INFO ("clean upload datas");
    if(isBlock(gid) != -1){
        clearBlockID(gid);
    }else{
        clearGroupData(gid);       
    }
    stMqttMsgInfo.iscoms = 0;
    setProfileConf(lastVideoSize,getPathDataSize(stMqttMsgInfo.baseUploadPath));
    return 0; 
}
INT32 mqttDataSyncDone (CHAR *buf, UINT32 len)
{
    CHAR  result[0x40] = {0};
    CHAR  groupID[0x40] = {0};
    CHAR  blockID[0x40] = {0}; 
    CHAR  dc[0x40] = {0}; 
    CHAR  sBuf[0x200] = {0};
    CHAR  count[0x40] = {0};
    CHAR  fSize[0x40] = {0};
    CHAR  eTime[0x40] = {0};
    UINT8 isblock = 1;
    INT32 timeout = 10;
    UINT32 waitTime = 0;

    if (DATA_SYNC_END_SUM == sscanf(buf, "%[^,],%[^,],%[^,],%[^,],%[^,],%[^,],%[^;]" ,count ,fSize ,eTime ,groupID ,blockID ,dc ,result)){
        if(strcmp(result,"complete") == 0){//rm file ,kill lftp
            waitTime = currentTimeGet_int();
            while(stMqttMsgInfo.dataSync == 1){
                usleep(200000);//200ms                  
                UINT32 curTime = currentTimeGet_int();       
                if((curTime - waitTime) >= timeout){
                    htkj_system("killall -9 lftp > /dev/null 2>&1");  
                    break;
                }  
            }        
            stMqttMsgInfo.iscoms = 1;
            if(atoi(dc) == 1){
                if(real_cleanData(groupID ,blockID) != 0){
                    strcpy(result,"ID error");
                }else{//clean ok
                    strcpy (result, "ok"); 
                    isblock = getGroupCount() > 0 ? 1 :0;
                    stMqttMsgInfo.iscoms = 0;
                }
            }else{                
                strcpy (result, "ok"); 
                isblock = getGroupCount() > 1 ? 1 :0; 
            }            
        }else{
            strcpy(result,"Parse error");
        }
    }else{
        strcpy(result,"Parse error");
    }
    stMqttMsgInfo.dataSync = 0;
    sprintf (sBuf, "%s,%s,%d,%s" ,groupID ,blockID ,isblock ,result);  

    return air_send_msg(CMD_UPLOAD_RET ,sBuf, strlen(sBuf), 1);
}

INT32 mqttCleanHdisk (CHAR *buf, UINT32 len)
{
    CHAR  result[0x40]  = {0};  
    CHAR  sBuf[0x100] = {0};   
    CHAR  groupID[0x40] = {0};
    CHAR  blockID[0x40] = {0};  

    strcpy (result, "error");
    if (CLEAN_HDISK_SUM == sscanf(buf, "%[^,],%[^;]", groupID, blockID))
    {
        if(real_cleanData(groupID ,blockID) != 0){
            strcpy(result,"ID error");
        }else{//clean ok
            strcpy (result, "ok"); 
        }         
    }else{
        strcpy(result,"Parse error");
    }

    sprintf (sBuf, "%s,%s,%s" ,groupID ,blockID ,result);  

    return air_send_msg(CMD_DISK_CLEAR ,sBuf, strlen(sBuf), 1);
}

INT32 verUpdate_task (void *arg)
{
    //to check the tmpVer

    if (strlen(stMqttMsgInfo.tmpVer) > 0){
	if (0 == strcmp(stMqttMsgInfo.tmpVer, stMqttMsgInfo.appVer)){//upgrade seccess
		INFO("upgrade success oldVer:%s,curVer:%s,appVer:%s,tmpVer%s",stMqttMsgInfo.oldVer,stMqttMsgInfo.curVer,stMqttMsgInfo.appVer,stMqttMsgInfo.tmpVer);		
		htkj_system("rm -fr %s/%s",stMqttMsgInfo.upgradeDir,stMqttMsgInfo.oldFilename);//rm old file		
		strcpy (g_upgradeRet, "complete"); 		
	} else {//upgrade failed			
		INFO("upgrade failed... oldVer:%s,curVer:%s,appVer:%s,tmpVer%s",stMqttMsgInfo.oldVer,stMqttMsgInfo.curVer,stMqttMsgInfo.appVer,stMqttMsgInfo.tmpVer);		
		htkj_system("rm -fr %s/%s",stMqttMsgInfo.upgradeDir,stMqttMsgInfo.tmpFilename);//rm new AIRFLASH_...	
		strcpy (g_upgradeRet, "failed");
	}
	//stMqttMsgInfo.upgradeEnd = 1;	
	mqttFirmwareUpdateDone (g_upgradeRet);
    }

    while (1)
    {
        sem_wait (&stMqttMsgInfo.upgradeSem);	
        verUpdate ();
    }
    return 0;
}
INT32 verUpdate (void)
{
    INT32 ret = 0;
    INT32 retVal = 0;

    CHAR result[0x100] = {0};
    CHAR cmd[0x200] = {0};
	
    /* 下载文件 */    
    INFO ("Load %s to %s Start ...",stMqttMsgInfo.upgradeDir,stVerUpdate.url);
    stMqttMsgInfo.isUpgrad = 1;
    ret = htkj_system ("wget -c -P %s %s >> %s 2>&1", stMqttMsgInfo.upgradeDir, stVerUpdate.url ,SHELL_LOG);
    if (0 != ret){
        Error ("Load fail, ret=%d!", ret);
        retVal = 1;

    } else {
    	//get updatepakge md5 sum value
	//Checking, is md5 sum value equie to the arguement
    	sprintf (cmd,"%s  %s", APP_MD5_CHECK , stVerUpdate.fileName); 	
	systemOut (cmd ,result, sizeof(result));
	
	if(0 != strcmp(result,"ok")){		
		htkj_system ("rm -f %s/%s", stMqttMsgInfo.upgradeDir,stVerUpdate.fileName);
	        htkj_system ("rm -fr %s/newsoft/",stMqttMsgInfo.upgradeDir);	
		Error ("md5sum Checking fail");
	        retVal = 2;
    	} else{
		//Download successfully 
		INFO ("Load Done.");
		setKeyValue_str ("app_tmp_version", stVerUpdate.version); 
		setKeyValue_str ("app_tmp_filename", stVerUpdate.fileName); 		
		//htkj_system ("%s %s &",APP_UPGRADE_SH, stVerUpdate.fileName);	
 		htkj_system ("cd %s/newsoft/ ; ./%s",stMqttMsgInfo.upgradeDir,APP_SETUP_FILE);	
	} 
    }
   
    if(retVal == 1){
	strcpy (g_upgradeRet, "down failed");
	mqttFirmwareUpdateDone(g_upgradeRet);
	stMqttMsgInfo.isUpgrad = 0;
	INFO("upgrade %s FAIL",stVerUpdate.version);
    }else if(retVal == 2){
	strcpy (g_upgradeRet, "md5 error");
	mqttFirmwareUpdateDone(g_upgradeRet);
	stMqttMsgInfo.isUpgrad = 0;
	INFO("upgrade %s FAIL",stVerUpdate.version);
    }
    
    return OK;
}

INT32 mqttFirmwareUpdate (CHAR *buf, UINT32 len)
{   
    CHAR result[0x40] = {0};
    INT32  ret = 0;   
  
    if(stMqttMsgInfo.isUpgrad == 1){
	return air_send_msg(CMD_UPGRADE_CMD ,"ok", strlen("ok"), 1);
    }
    memset (stVerUpdate.fileName, 0x00, sizeof(stVerUpdate.fileName));
    memset (stVerUpdate.version, 0x00, sizeof(stVerUpdate.version));
    memset (stVerUpdate.md5, 0x00, sizeof(stVerUpdate.md5));
    memset (stVerUpdate.url, 0x00, sizeof(stVerUpdate.url));
    if (FIRMWARE_UPDATE_SUM == sscanf(buf, "%[^,],%[^,],%[^,],%[^;]",  stVerUpdate.fileName, stVerUpdate.version, stVerUpdate.md5, stVerUpdate.url))
    {
	
        INFO("filename:%s,version:%s,MD5:%s,url:%s\n",stVerUpdate.fileName, stVerUpdate.version, stVerUpdate.md5, stVerUpdate.url);
        if (NULL != strrchr (stVerUpdate.url, '/')) {
            strcpy (result, "ok");           
        } else {
            strcpy (result, "error");
        }
    } else{
        strcpy (result, "error");
    }
    ret = air_send_msg(CMD_UPGRADE_CMD ,result, strlen(result), 1);
   
    if (0 == strcmp (result, "ok")){
        // 启动升级任务
	sem_post (&stMqttMsgInfo.upgradeSem);
    }

    return (ret);
}

INT32 mqttFirmwareUpdateDone (CHAR *sBuf)
{
    INT32  ret = 0; 

    ret = air_send_msg(CMD_UPGRADE_RET ,sBuf, strlen(sBuf), 1);
    return (ret);
}

INT32 mqttFirmwareUpdateDone_ACK (CHAR *buf, UINT32 len)
{
    CHAR result[0x40] = {0};
    INT32  ret = 0;
    
    if (DATA_SYNC_START_SUM == sscanf(buf, "%[^;]", result))
    {
    }
    //stMqttMsgInfo.upgradeEnd = 0;
    if(0 == strcmp(g_upgradeRet, "failed")){
	rmKeyValue("app_tmp_version");
	rmKeyValue("app_tmp_filename");
	htkj_system("%s %s",APP_ROLLBACK_SH,stMqttMsgInfo.tmpFilename);//rollback app
		
    }else if(0 == strcmp(g_upgradeRet,"complete")){
	setKeyValue_str("app_old_version", stMqttMsgInfo.curVer);
	setKeyValue_str("app_old_filename", stMqttMsgInfo.curFilename);
	setKeyValue_str("app_cur_version", stMqttMsgInfo.tmpVer);		
	setKeyValue_str("app_cur_filename", stMqttMsgInfo.tmpFilename);
	rmKeyValue("app_tmp_version");
	rmKeyValue("app_tmp_filename");	
    }
    return (ret);
}

INT32 getGroupDataSize()
{
    UINT32 uDisk = 0;
    INT32 index[BLOCK_MAX_COUNT] = {0};
    CHAR path[1024];
    INT32 size = 0;
    INT32 i = 0;

    size = sort_dir(stMqttMsgInfo.baseUploadPath,VIDEO_GROUP ,index);
    
    if(size <= 0){
        return 0;
    }
    for(i = 0; i < size;i++){
        sprintf(path,"%s/%s%d" ,stMqttMsgInfo.baseUploadPath ,VIDEO_GROUP ,index[i]);
        uDisk += getPathDataSize(path);
    }
    
    return uDisk;
    
}
INT32 hbInfoGet (CHAR *disk_used, CHAR *progress)
{   
    double g = 0;
    UINT32 useDisk = 0;  

    useDisk = getGroupDataSize();

    g = (double)useDisk/1024/1024;//GB
    sprintf (disk_used, "%.2lf", g);
   
    strcpy (progress, "0");

  //  getSpeed(speed); 
    return OK;
}

INT32 mqtt_HB_Msg (void)
{    
    CHAR disk_used[0x40] = {0};
    CHAR progress[0x40] = {0}; 
    CHAR   sBuf[0x80] = {0};
    //INT32 speed = 0;

    stMqttMsgInfo.speed = getSpeed();

    hbInfoGet (disk_used, progress);
    
    sprintf (sBuf, "%s,%s,%s,%.2lf", stMqttMsgInfo.diskSize, disk_used, progress, stMqttMsgInfo.speed/1024.0);
      
    return air_send_msg(CMD_HEART_BEAT ,sBuf, strlen(sBuf), 0);
}
INT32 mqttHeartBeat (void)
{
    mqtt_HB_Msg ();   
    return OK;
}

INT32 mqtt_HB_MSG_ACK (CHAR *buf, UINT32 len)
{  
    CHAR result[0x40] = {0};
    CHAR percent[0x40]={0};

    //if (HB_MSG_ACK_SUM == sscanf(buf, "%[^;]", result))
    if (HB_MSG_ACK_SUM == sscanf(buf, "%[^,],%[^;]", percent,result))
    {
        
    }
    stMqttMsgInfo.percent = atoi(percent);
    stMqttMsgInfo.HBAckTime = currentTimeGet_int();
   
    return OK;
}
UINT8 load_6ATime(void)//settime
{
    UINT32 _6aTi = file_lastTime(A6_PATH);
    UINT32 localTi = currentTimeGet_int();
    INT8 _6aTs[0x40] = {0};
    INT8 localTs[0x40] = {0};
    INT8 startTs[0x40] = {0};

     if(_6aTi <= 0)
        return 0;
        
    _6aTi -= 28800;//8*60*60
    
    if(_6aTi > ((localTi/86400 + 3) * 86400)){//set 6A time > 2019-10-28 0:0:0
        //_6aTi -= 28800;//8*60*60
        intTime2Str(_6aTi ,_6aTs);
        intTime2Str(localTi ,localTs);
        htkj_system ("date -s '%s'", _6aTs);
        stMqttMsgInfo.powerONTime = _6aTi - (localTi - stMqttMsgInfo.powerONTime);        
        intTime2Str(stMqttMsgInfo.powerONTime ,startTs);
        localTi = currentTimeGet_int();

        INFO("load 6A time:[%s;%d;%s=%s]",startTs,localTi - stMqttMsgInfo.powerONTime,localTs ,_6aTs);
       
        if(getProfileConf(lastNewGroupTime) == 0){//init data/.profile
            setProfileConf(lastNewGroupTime ,currentTimeGet_int());
        }
        return 1;
    }    
    if(access(REGSub6ATimeFlg,F_OK) != 0){
        if( localTi > 0 &&  (localTi > _6aTi) && ((localTi - _6aTi) > 10) &&  ((localTi - _6aTi) < 1800)){
            htkj_system("echo regSub6ATime=%d > %s",localTi - _6aTi + 15,REGSub6ATimeFlg);        
            INFO("no need load 6A time:[%d > %d]" ,localTi ,_6aTi);
            return 1;
        }
    }
    if(_6aTi < localTi)
        return 1;
    return 0;
}
void loop_airLOG(void)
{
   UINT64 _6ASize = getPathDataSize(A6_PATH);//,"/tmp/.h6pathsize.tmp");
   UINT64 uploadSize = getPathDataSize(stMqttMsgInfo.baseUploadPath) + stMqttMsgInfo.videoSize;//getPathDataSize(stMqttMsgInfo.uploadLinkPath);
   INT8 buf[0x40] = {0};
   INT32 videoSize = getPathDataSize(stMqttMsgInfo.transferPath);
   
   if(videoSize < 0)
	videoSize = 0;
    
   INFO("loopLog:%s;time[%d-%d];data[%d-%d];video[%d];6A[%d];UTimes[%d-%d];isUpan[%d]",
                                intTime2Str(stMqttMsgInfo.powerONTime,buf)//power on datetime				
                                ,stMqttMsgInfo.oNTimeDuration//since reg
                                ,currentTimeGet_int()-stMqttMsgInfo.powerONTime	//since  power on 
				,stMqttMsgInfo.uploadStartSize	//when  power on 	
				,uploadSize
                                ,videoSize			
				,_6ASize
				,stMqttMsgInfo.UplugTimes + getUSBDiskPlugTimes(0) //since reg
                                ,getUSBDiskPlugTimes(2)//since power on or upload
				,getUSBDisk()//if usbdisk	
				);  
}
void log_powerOnTime(void)
{         
    stMqttMsgInfo.oNTimeDuration += 60; 
    setProfileConf(lastPowerTimeLen ,stMqttMsgInfo.oNTimeDuration);
    htkj_system("echo \"0,%d,1,%d\" > %s" , stMqttMsgInfo.powerONTime ,currentTimeGet_int() ,ONOFF_timeProfile);   
}

INT32 check_onoffTimes(void)
{
    INT8 sBuf[MSG_QUEUE_DATA_LEN] = {0};
    FILE* fp;
    INT8 line[LINE_CONTENT_MAX_LEN];
    INT32 i = 0;
    INT32 len = 0;
   
    if(access("/tmp/.onoffTimes.tmp" , F_OK) == 0)
        return ERROR;   
    if(access(ONOFF_timesFile,F_OK) == 0){
        fp = fopen(ONOFF_timesFile, "r");
        if(fp != NULL){
            while(feof(fp) == 0){
                memset(line, 0, sizeof(line));
                fgets(line, sizeof(line), fp);
                for(i = 0; i < sizeof(line); i++){
                    if((line[i] == '\0') || (line[i] == '\r') || (line[i] == '\n')){
                        line[i] = '\0';
                        break;
                    }
                }
                if(strlen(line) < 25)
                    continue;
                if((len + strlen(line) + 1) < sizeof(sBuf))
                    len += sprintf(&sBuf[len],"%s,",line);  
            }
            fclose(fp);
        }else{
            Error("%s: GR Opent file %s failed.", __FILE__, ONOFF_timesFile);
        }
    }
    htkj_system("rm -fr %s > /dev/null 2>&1",ONOFF_timesFile);
    htkj_system("echo 1 > /tmp/.onoffTimes.tmp");

    if((len + 25) < sizeof(sBuf)){
        len += sprintf(&sBuf[len] , "0,%d,1,%d" ,stMqttMsgInfo.powerONTime ,currentTimeGet_int());
    }else{
        len--;
    }

    return air_send_msg(CMD_ONOFF ,sBuf, len, 0);
}
INT32 onoffTime_ACK(CHAR *buf, UINT32 len)
{
    CHAR result[0x40] = {0};
    
    if (ONOFF_MSG_ACK_SUM == sscanf(buf, "%[^;]", result))
    {
    }
    return OK;
}
UINT8 isDivideGroupTime()
{
    UINT32 lastTi = getProfileConf(lastNewGroupTime);
    UINT32 curTi = currentTimeGet_int();

    if((lastTi == 0) && ((curTi - stMqttMsgInfo.powerONTime) > (stMqttMsgInfo.route_min_time*60))){//2019-10-25 19:40:00 sytem no time start
        return 1;
    }else if((lastTi > 0)){
        if((curTi > lastTi) && ((curTi - lastTi) > (stMqttMsgInfo.route_min_time*60))){
             return 1;
        }else if((curTi <= lastTi) && ((curTi - stMqttMsgInfo.powerONTime) > (stMqttMsgInfo.route_min_time*60))){
              return 1;
        }  
    }
    return 0;
}
void check_poe(void)
{
    static UINT32 count = 0;
    CHAR noNeed = 0;
    double readySize = (double)stMqttMsgInfo.videoSize/1024/1024;//upload/dataSet
    double linkReadySize = (double)getPathDataSize(stMqttMsgInfo.baseUploadPath)/1024/1024;//upload/data

    if((stMqttMsgInfo.regHostAck == MQTT_REG_HOST_MSG_OK) && (isGroup() < 0) && (stMqttMsgInfo.isHaveFwUpgrade == 0) && (isDivideGroupTime() <= 0)){//online,no group,no fwUpgrade,no split group 
        noNeed = 1;
    } else {
        noNeed = 0;
    }
    if(noNeed > 0 && stMqttMsgInfo.poeIsON > 0){//close case
        count++;
        if(count > 5){//wait 5s close poe
            setPOE(0);             
            count = 0;
        }
        return;
    }
//need open
    count = 0;
    if(stMqttMsgInfo.poeIsON){//poe opened && no need close	  
        return;
    }
    if(isDivideGroupTime() > 0 || isGroup() >= 0 || (linkReadySize + readySize) >= stMqttMsgInfo.block_max_size){//have group or >30
        setPOE(1);  
    } 
}
void check_divideBlock(void)
{
    double size = 0;  
     
    pthread_mutex_lock(&stMqttMsgInfo.block_mutex);
    size = stMqttMsgInfo.videoSize;//getPathDataSize(stMqttMsgInfo.uploadLinkPath)/1024*1.0/1024.0;

    if(size > stMqttMsgInfo.block_max_size*1024*1024){
        divide_block();	
    }
    pthread_mutex_unlock(&stMqttMsgInfo.block_mutex);
}
INT32 mqttMsgReport_task (void *arg)
{
    UINT8 setA6Time = 0;  
    UINT8 onoff = 0;
    UINT32 cTime = 0; 
    UINT32 regTime = 0;
    UINT32 regTimes = 0;
    UINT32 hbTime = 0;   
    UINT32 logTime = 0;
    UINT32 powerTime = 0;
    UINT32 poeTime = 0;

    while (1)
    {
        if(stMqttMsgInfo.timeChange){
            hbTime = 0; 
            regTime = 0;  
            logTime = 0;
            powerTime = 0;
            stMqttMsgInfo.HBAckTime = currentTimeGet_int();
            stMqttMsgInfo.timeChange = 0;
        }
        usleep(500000);//500ms
        cTime = currentTimeGet_int();
        if((stMqttMsgInfo.regHostAck != MQTT_REG_HOST_MSG_OK) && (stMqttMsgInfo.link == IOT_STATUS_CONNECT)){
            if((cTime - regTime) >= stMqttMsgInfo.regInterval){
                if( regTimes >= 3 ){
                    Error ("register failed 3 times, SOCKET IOT_STATUS_DROP !");
                    htkj_system("rm -f %s" ,BOND_CHECK_ENABLE);
                    stMqttMsgInfo.link = IOT_STATUS_DROP;                    
                    regTimes = 0;
	        }else{
                    mqttMsgRegHostRule ();//send reg
                    regTimes++;
                    regTime = cTime;
                }
            }
        }

        if (stMqttMsgInfo.regHostAck == MQTT_REG_HOST_MSG_OK)
        {
            htkj_system("touch %s" ,BOND_CHECK_ENABLE);
            cTime = currentTimeGet_int();
            if ( regTimes > 0 )
            {               
                stMqttMsgInfo.HBAckTime = cTime;		
            }

            regTimes = 0;
            if( (cTime - stMqttMsgInfo.HBAckTime) >= stMqttMsgInfo.HBTimeout )
            {
                Error ("HeartBeat Timeout, SOCKET IOT_STATUS_DROP !");
                htkj_system("rm -f %s" ,BOND_CHECK_ENABLE);
                stMqttMsgInfo.regHostAck = MQTT_REG_HOST_MSG_INIT;               
                msgBdClean();
                stMqttMsgInfo.link = IOT_STATUS_DROP;
            }else if(stMqttMsgInfo.dataSync){
                if((cTime - hbTime) >= stMqttMsgInfo.HBInterval){//uploading per HBInterval
		    mqttHeartBeat();
                    hbTime = cTime;
                }
            }else if((cTime - hbTime) >= stMqttMsgInfo.HBTimeout/3){//no uploading per 10s
                mqttHeartBeat();
                hbTime = cTime;                
            }          
            if(onoff == 0){
                onoff = check_onoffTimes() == OK ? 1 : 0;
            }
            
        }

        if(stMqttMsgInfo.isUsbCollect){
            if(!setA6Time){
                setA6Time = load_6ATime();
            }  
        } 
        if((cTime - logTime) >= stMqttMsgInfo.loopLogInterval){
            loop_airLOG(); 
            logTime = cTime;
			}
        if((cTime - powerTime) >= (stMqttMsgInfo.HBTimeout*2)){
            log_powerOnTime(); 
            check_divideBlock();
            powerTime = cTime;
        }
        if((cTime - poeTime) > 0){
            check_poe();
            poeTime = cTime;
        }
    }
    return 0;
}
/*******************************************************************************
*   函数名称: msgCmdRun
*   功    能: 执行命令
*   函数类型:
*   参    数:
*   参数名称       类  型                   输入/输出         描述
*
*   函数返回: OK 或失败的错误码
*   说    明:
*******************************************************************************/
INT32 mqttMsgCmdRun (INT32 type, CHAR *data, UINT32 len)
{
    UINT32 ix = 0;
    INT32  ret = 0;
    MQTT_MSG_PROCESS_CMD_T *stMsgCmdList;

    stMsgCmdList = g_stMsgCmdList;

    while (1)
    {
        if (type ==  stMsgCmdList[ix].key){
            if (NULL != stMsgCmdList[ix].fun){
               // INFO ("%s", stMsgCmdList[ix].desc);
                ret = stMsgCmdList[ix].fun (data, len);		
                return (ret);
            }
        }
        ix ++;
    }

    //return OK;
}


/******************************* 源文件结束 **********************************/
